package kafka2

import (
	"context"
	"errors"
	"fmt"
	"gitee.com/chejiangyi/bsfgo/core/base2"
	"gitee.com/chejiangyi/bsfgo/core/utils"
	"github.com/IBM/sarama"
	"strings"
	"time"
)

type KafkaMessage struct {
	Key, Value []byte
	Topic      string
	Partition  int32
	Offset     int64
	Timestamp  time.Time
	Session    sarama.ConsumerGroupSession
}
type KafkaConsumer struct {
	ConsumerGroup sarama.ConsumerGroup
}

// Consume 开始消费指定主题
func (k *KafkaConsumer) Consume(topics []string, consumeFuc func(message *KafkaMessage, event string)) {
	if len(topics) == 0 {
		panic(errors.New("消费者topic未定义"))
	}

	if err := k.ConsumerGroup.Consume(context.Background(), topics, &KafkaConsumerGroupHandler{Consume: consumeFuc, Topic: topics}); err != nil {
		panic(base2.NewBsfError2(fmt.Sprintf("topics:%v,消费异常", strings.Join(topics, ",")), err))
	}
}

func (k *KafkaConsumer) Close() {
	if k.ConsumerGroup != nil {
		k.ConsumerGroup.Close()
	}
}

type KafkaConsumerGroupHandler struct {
	Topic   []string
	Consume func(message *KafkaMessage, event string)
}

func (h *KafkaConsumerGroupHandler) Setup(session sarama.ConsumerGroupSession) error {
	utils.LogUtil.Info(fmt.Sprintf("kafka 消费开启,topic:%v", strings.Join(h.Topic, ",")))
	h.Consume(&KafkaMessage{Session: session}, "setup")
	return nil
}

func (h *KafkaConsumerGroupHandler) Cleanup(session sarama.ConsumerGroupSession) error {
	utils.LogUtil.Info(fmt.Sprintf("kafka 消费关闭,topic:%v", strings.Join(h.Topic, ",")))
	h.Consume(&KafkaMessage{Session: session}, "cleanup")
	return nil
}

func (h *KafkaConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		h.Consume(&KafkaMessage{Topic: message.Topic, Key: message.Key, Value: message.Value,
			Partition: message.Partition, Offset: message.Offset, Session: session}, "message")
		session.MarkMessage(message, "")
		if !base2.BsfConfigInstance.GetKafkaConsumerOffsetsAutoCommitEnable() {
			session.Commit()
		}
	}
	return nil
}
